package com.example.demo.rocketmq.listener;

import com.example.demo.rocketmq.provider.service.CommonProviderService;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.messaging.Message;

import java.lang.reflect.InvocationTargetException;
import java.util.List;

@RocketMQTransactionListener
public class MqTxLocalListener implements RocketMQLocalTransactionListener {

    final
    List<CommonProviderService> commonProviderServices;

    public MqTxLocalListener(List<CommonProviderService> commonProviderServices) {
        this.commonProviderServices = commonProviderServices;
    }

    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        commonProviderServices.forEach(commonService -> {
            if (message.getHeaders().get("beanName").equals(commonService.getBeanName())) {
                try {
                    String methodName = (String) message.getHeaders().get("methodName");
                    String parameterType = (String) message.getHeaders().get("parameterTypes");
                    commonService.localTransactionExeute(commonService, methodName, Class.forName(parameterType), o);
                } catch (NoSuchMethodException | InvocationTargetException | IllegalAccessException |
                         ClassNotFoundException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        return RocketMQLocalTransactionState.COMMIT;
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        //事务回查, 默认30s一次 , 15次内为返回COMMIT，则认为事务失败
        return null;
    }
}
